package com.jeffse.light.stream.core.data.stream;

import com.jeffse.light.stream.core.constans.OutTagConstans;
import com.jeffse.light.stream.core.data.context.Context;
import com.jeffse.light.stream.core.data.context.OperatorContext;
import com.jeffse.light.stream.core.data.descriptor.operatorDescriptor.OperatorDescriptor;
import com.jeffse.light.stream.core.data.operator.OutPutStreamOperator;
import com.jeffse.light.stream.core.data.operator.StreamOperator;
import com.jeffse.light.stream.core.function.advance.FlatMapFunction;
import com.jeffse.light.stream.core.function.advance.MapFunction;
import com.jeffse.light.stream.core.function.advance.SlideWindowFunction;
import com.jeffse.light.stream.core.function.advance.WindowFunction;
import com.jeffse.light.stream.core.function.base.*;
import com.jeffse.light.stream.core.pojo.OutTag;
import com.jeffse.light.stream.core.selector.KeyedSelector;
import com.jeffse.light.stream.core.data.context.SourceContext;
import com.jeffse.light.stream.core.thread.LightStreamThread;
import com.jeffse.light.stream.core.utils.ThreadUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.logging.log4j.ThreadContext;

import java.util.HashMap;
import java.util.Map;

@Slf4j
public  class DataStream<KEY, T, V> implements OperatorContext<KEY, T, V>, Context<KEY, T, V>, AbstractDataStream<KEY, T, V> {
    static final long serialVersionUID =-5666638870709238304L;

    private String dataStreamName = "";

    /**
     * 数据源
     */
    SourceFunction sourceFunction = null;

    /**
     * 数据源上下文
     */
    SourceContext sourceContext = null;

    /**
     * 上下文
     */
    Context ctx = null;

    private OutPutStreamOperator<KEY, T,V> streamOperator;

    /**
     * 所有 outTag 的 map 集合
     */
    private Map<String, OutTag> outTagMap = new HashMap<>();

    /**
     * 所有 StreamOperator 的 map 集合
     */
    private Map<String, OutPutStreamOperator> streamOperatorMap = new HashMap<>();

    private SinkFunction sinkFunction = null;

    /**
     * 是否已经启动
     */
    protected Boolean isRunning = false;

    public DataStream(){
        init();
    }

    public DataStream(String dataStreamName){
        this.dataStreamName = dataStreamName;
        ThreadContext.put("ROUTINGKEY", dataStreamName);
        init();
    }

    /**
     * 初始化
     */
    private void init(){
        ctx = this;
        OutTag outTag = new OutTag(OutTagConstans.MAIN);
        outTagMap.put(OutTagConstans.MAIN, outTag);
        streamOperator = new OutPutStreamOperator<>(this);
        streamOperator.setOutTag(outTag);
        streamOperatorMap.put(OutTagConstans.MAIN, streamOperator);
    }

    @Override
    public String getDataStreamName(){
        return dataStreamName;
    }

    @Override
    public StreamOperator getStreamOperator() {
        return streamOperator;
    }

    @Override
    public DataStream processElement(Object value, OperatorDescriptor operatorDescriptor) throws Exception {
        streamOperator.processElement(value, operatorDescriptor);
        return this;
    }

    /**
     * 添加数据源
     * @param sourceFunction
     * @return
     */
    @Override
    public DataStream addSource(SourceFunction sourceFunction)  {
        this.sourceFunction = sourceFunction;
        sourceContext = new SourceContext() {
            @Override
            public void collect(Object element) throws Exception {
                ctx.collect(element, streamOperatorMap.get(OutTagConstans.MAIN).getFirstDescriptor(), outTagMap.get(OutTagConstans.MAIN));
            }
        };
        return this;
    }

    public SourceFunction<T> getSourceFunction(){
        return sourceFunction;
    }

    /**
     * 添加输出
     * @param sinkFunction
     * @return
     */
    @Override
    public DataStream addSink(SinkFunction sinkFunction) {
        this.sinkFunction = sinkFunction;
//        streamOperator.addSink(sinkFunction);
        return this;
    }

    @Override
    public DataStream map(MapFunction mapFunction) {
        streamOperator.map(mapFunction);
        return this;
    }

    @Override
    public DataStream flatMap(FlatMapFunction flatMapFunction){
        streamOperator.flatMap(flatMapFunction);
        return this;
    }

    @Override
    public DataStream keyedBy(KeyedSelector<T, KEY> keyedSelector) {
        streamOperator.keyedBy(keyedSelector);
        return this;
    }

    @Override
    public DataStream setWaterMark(WaterMarkFunction<T> waterMarkFunction) {
        streamOperator.setWaterMark(waterMarkFunction);
        return this;
    }

    @Override
    public DataStream union(OutTag outTag) {
        streamOperator.union(outTag);
        return this;
    }

    @Override
    public DataStream window(Enum<WindowType> windowType, Integer num, Enum<UnitType> unit, Boolean isEventTime) throws Exception {
        streamOperator.window(windowType, num, unit, isEventTime);
        return this;
    }

    @Override
    public DataStream slideWindow(Enum<WindowType> windowType, Integer x, Enum<UnitType> unitX, Integer y, Enum<UnitType> unitY, Boolean isEventTime) throws Exception {
        streamOperator.slideWindow(windowType, x, unitX, y, unitY, isEventTime);
        return this;
    }

    @Override
    public Integer count() {
        return null;
    }

    @Override
    public Double avg() {
        return null;
    }

    @Override
    public Double sum() {
        return null;
    }

    @Override
    public Double max() {
        return null;
    }

    @Override
    public Double min() {
        return null;
    }

    @Override
    public void doSink(Object value) throws Exception {
        sinkFunction.run(value, ctx);
    }

    /**
     * 数据源的输出
     * @param element
     * @throws Exception
     */
    @Override
    public void collect(Object element, OperatorDescriptor operatorDescriptor, OutTag outTag) throws Exception {
        streamOperator = streamOperatorMap.get(outTag.getName());
        streamOperator.processElement(element, operatorDescriptor);
    }

    @Override
    public void collect(Object element, OutTag outTag) throws Exception {
        streamOperator = streamOperatorMap.get(outTag.getName());
        streamOperator.processElement(element, streamOperator.getFirstDescriptor());
    }

    @Override
    public void collect(Object element, OperatorDescriptor operatorDescriptor) throws Exception {
        streamOperator.processElement(element, operatorDescriptor);
    }

    /**
     * 添加过程函数
     * @param abstractProcessFunction
     * @return
     */
    @Override
    public DataStream process(AbstractProcessFunction abstractProcessFunction) throws Exception {
        streamOperator.process(abstractProcessFunction);
        return this;
    }

    @Override
    public DataStream process(WindowFunction windowFunction) throws Exception {
        streamOperator.process(windowFunction);
        return this;
    }

    @Override
    public DataStream process(SlideWindowFunction slideWindowFunction) throws Exception {
        streamOperator.process(slideWindowFunction);
        return this;
    }

    @Override
    public OutPutStreamOperator getSideOutput(OutTag outTag){
        OutPutStreamOperator streamOperator = new OutPutStreamOperator<KEY, T, V>(this);
        streamOperator.setOutTag(outTag);
        streamOperatorMap.put(outTag.getName(), streamOperator);
        return streamOperator;
    }

    @Override
    public void union(OutTag sourceTag, OutTag targetTag, OutTag unionTag, OutPutStreamOperator outPutStreamOperator) {
        streamOperatorMap.get(sourceTag.getName()).setSinkTag(unionTag);
        streamOperatorMap.get(targetTag.getName()).setSinkTag(unionTag);
        streamOperatorMap.put(unionTag.getName(), outPutStreamOperator);
    }

    @Override
    public void start() throws Exception {
        isRunning = true;
        ThreadUtils.execute(new LightStreamThread(()->{
            ThreadContext.put("ROUTINGKEY", dataStreamName);
            try {
                sourceFunction.run(sourceContext);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, dataStreamName), dataStreamName);
    }

    @Override
    public void stop() throws Exception {
        isRunning = false;
        if(sourceFunction instanceof AbstractSourceFunction){
            ((AbstractSourceFunction) sourceFunction).cancel();
        }
    }

    @Override
    public void run() throws Exception {
        isRunning = true;
        sourceFunction.run(sourceContext);
    }

    @Override
    public Boolean getIsRunning(){
        return isRunning;
    }
}
